package com.example.mq.kafka.consumer;

import com.example.mq.kafka.dataobject.Kafka01Message;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

/**
 * @author luowj12
 * @since 2021/4/25 10:10
 */
@Component
@Slf4j
public class Kafka01Consumer {


    @KafkaListener(topics = Kafka01Message.TOPIC,concurrency="2",
            groupId = "demo01-consumer-group-" + Kafka01Message.TOPIC)
    public void onMessage(ConsumerRecord<String, Kafka01Message> record) {
        Kafka01Message message = record.value();
        int partition = record.partition(); // 获取消息的 partition
        log.info("[onMessage][线程编号:{},消息内容:{} ,消息所属的partition:{}]", Thread.currentThread().getId(), message,partition);
    }

}
